package org.example.flinketl.util;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.example.flinketl.vo.*;

import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
import scala.collection.immutable.HashSet;

public class MyConfig {
  // clickhouse
  public final static String clickhouseUrl = "jdbc:clickhouse://127.0.0.1:8123/dm";
  public final static String databaseName = "dm";
  public final static String username = "default";
  public final static String password = "root";
  public final static Class<?>[] classes =  new Class[] {
      org.example.flinketl.vo.ContractMx.class,
      org.example.flinketl.vo.DjkMx.class,
      org.example.flinketl.vo.DsfMx.class,
      org.example.flinketl.vo.DuebillMx.class,
      org.example.flinketl.vo.EtcMx.class,
      org.example.flinketl.vo.GrwyMx.class,
      org.example.flinketl.vo.GzdfMx.class,
      org.example.flinketl.vo.HuanbMx.class,
      org.example.flinketl.vo.HuanxMx.class,
      org.example.flinketl.vo.SaMx.class,
      org.example.flinketl.vo.SbybMx.class,
      org.example.flinketl.vo.SdrqMx.class,
      org.example.flinketl.vo.ShopMx.class,
      org.example.flinketl.vo.SjyhMx.class
  };

  // table
  public final static Map<Class<?>, LinkedHashSet<String>> fields = initFields();
  public final static ClickHouseDataSource dataSource = initClickHouseDataSource();
  public final static Map<String, Class<?>> tableTye = initTableType();
  public final static Map<Class<?>, String> sqls = initSqls();
  public final static Map<Class<?>, List<Method>> methods = initMethods();

  // kafka
  public final static String groupId = "test";
  public final static String kafkaUrl = "127.0.0.1:9092";
  public final static String topic = "0401";

  private static Map<Class<?>, LinkedHashSet<String>>  initFields() {

    Map<Class<?>, LinkedHashSet<String>> fields = new HashMap<>();
    for (Class<?> t : classes) {
      LinkedHashSet<String> keys = new LinkedHashSet<>();
      for (Field field : t.getDeclaredFields()) {
        keys.add(field.getName());
      }
      fields.put(t, keys);
    }
    return fields;

  }

  private static ClickHouseDataSource initClickHouseDataSource(){
    ClickHouseProperties properties = new ClickHouseProperties();
    properties.setUser(username);
    properties.setPassword(password);
    ClickHouseDataSource tmp = new ClickHouseDataSource(clickhouseUrl, properties);
    System.out.println("datasource");
    ClickHouseConnection con = null;
    try {
      con = tmp.getConnection();
    } catch (SQLException e) {
      // TODO Auto-generated catch block
      System.out.println("fail to create datasource connection");
      e.printStackTrace();
    }
    if(con == null)System.out.println("con from datasource is null");
    return tmp;
  }

  private static Map<String, Class<?>> initTableType(){
    Map<String, Class<?>> map =  new HashMap<>();
    for(Class<?> cl : classes){
      String className = cl.getSimpleName();
      String eventType =  className.substring(0, className.length()-2).toLowerCase();
      map.put(eventType, cl);
    }
    return map;
  }

  private static Map<Class<?>, String> initSqls(){
    Map<Class<?>, String> map = new HashMap<>();
    fields.forEach((key,value)->{
      // Method method = key.
      Method method;
      String tableName;
      try {
        method = key.getMethod("getTableName");
        tableName = (String) method.invoke(null);
      } catch (NoSuchMethodException | SecurityException | IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return;
      }
      StringBuilder sbField = new StringBuilder();
      StringBuilder sbValue = new StringBuilder();
      value.forEach((field)->{
        sbField.append(field).append(',');
        sbValue.append('?').append(',');
      });
      sbField.deleteCharAt(sbField.length()-1);
      sbValue.deleteCharAt(sbValue.length()-1);
      String sql = "INSERT INTO " + databaseName + "." + tableName + "(" + sbField.toString() + ")" + "VALUES(" + sbValue.toString() + ");";
      map.put(key, sql);
    });
    return map;
  }

  private static Map<Class<?>, List<Method>> initMethods(){
    Map<Class<?>, List<Method>> map = new HashMap<>();
    fields.forEach((key, value)->{
      List<Method> methodsList = new ArrayList<>();
      value.forEach(item->{
        StringBuilder methodName = new StringBuilder();
        methodName.append("get")
        .append(Character.toUpperCase(item.charAt(0)))
        .append(item.substring(1));
        Method method;
        try {
          method = key.getMethod(methodName.toString());
        } catch (NoSuchMethodException | SecurityException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
          return;
        }
        methodsList.add(method);
      });
      map.put(key, methodsList);
    });
    return map;
  }
}
